package watcher

import (
	"bufio"
	"encoding/json"
	"errors"
	"fmt"
	"io"
	"io/ioutil"
	"net"
	"net/url"
	"os"
	"os/exec"
	"os/user"
	"regexp"
	"strconv"
	"strings"
	"sync"
	"time"

	pathpkg "path"

	"../pb"
	"../utils"
	grpclb "github.com/liyue201/grpc-lb"
	"github.com/op/go-logging"
	"golang.org/x/net/context"
	"google.golang.org/grpc"
	"google.golang.org/grpc/naming"
)

func checkErr(err error) {
	if err != nil {
		log.Error(err)
		os.Exit(2)
	}
}

var log = logging.MustGetLogger("watcher_service")

type ProcWatcherService struct {
	mu              sync.RWMutex
	procs           map[string]*ProcessItem
	RefreshInterval int64
	defineFile      string
	lastScanTime    int64
	stopping        chan interface{}
	stopped         chan interface{}
	tkvalid         map[string]int64
	rpvalid         map[string][]string
	auth            string
	value           map[string]*pwMetric
}

type pwMetric struct {
	stat     *pb.Status
	duration int64
}

func (pws *ProcWatcherService) SetAuth(auth string) {
	if auth != "" {
		log.Infof("set auth host on: %v", auth)
		pws.auth = auth
	}
}

func NewProcWatcherService(fn string, interval int64) (*ProcWatcherService, error) {
	f, err := os.Stat(fn)
	checkErr(err)
	if !f.IsDir() {
		b, err := ioutil.ReadFile(fn)
		checkErr(err)
		p := &ProcWatcherService{}
		err = json.Unmarshal(b, &p.procs)
		checkErr(err)
		p.defineFile = fn
		p.RefreshInterval = interval
		return p, nil
	} else {
		files, err := ioutil.ReadDir(fn)
		checkErr(err)
		p := &ProcWatcherService{}
		for _, f := range files {
			if f.Name() != "mas.json" {
				path := pathpkg.Join(fn, f.Name())
				b, err := ioutil.ReadFile(path)
				checkErr(err)
				procs := make(map[string]*ProcessItem)
				err = json.Unmarshal(b, &procs)
				checkErr(err)
				if p.procs == nil {
					p.procs = procs
				} else {
					for k, v := range procs {
						p.procs[k] = v
					}
				}
			}
		}
		p.defineFile = fn
		p.RefreshInterval = interval
		return p, nil
	}
}

type ProcessItem struct {
	MatchMulti     bool   `json:"match_multi"`
	ProcessPattern string `json:"pattern"`
	ProcessUser    string `json:"user"`
	//	Become         string              `json:"become"`
	Commands map[string][]string `json:"commands"` //with commands params
	Tags     map[string]string   `json:"tags"`     //with tags params
	reg      *regexp.Regexp
	process  Process
}

func (pi *ProcessItem) Reset() {
	if pi.MatchMulti {
		pi.process = &MultiProcess{}
	} else {
		pi.process = &SingleProcess{}
	}
}

type Process interface {
	Status(string) (*pb.Status, error)
	Add(user.User, int64, int) error
	Info(bool) string
	Uptime() int64
}

func (pws *ProcWatcherService) Auth(ctx context.Context, token *pb.Token, rp string) (bool, error) {
	uname := token.GetUser()
	if pws.tkvalid == nil {
		pws.tkvalid = make(map[string]int64)
	}
	if pws.rpvalid == nil {
		pws.rpvalid = make(map[string][]string)
	}
	if pws.tkvalid[uname] >= time.Now().UnixNano()/1e6 {
		ok := false
		for _, urp := range pws.rpvalid[uname] {
			if urp == rp {
				ok = true
				break
			}
		}
		if ok {
			return true, nil
		}
	} else {
		pws.rpvalid[uname] = nil
	}

	if pws.auth != "" {
		ch := make(chan *naming.Update, 1)
		re := utils.NewResolver(ch)
		b := grpclb.NewBalancer(re, grpclb.NewRoundRobinSelector())
		ctx, _ := context.WithTimeout(context.Background(), 10*time.Second)
		conn, connerr := grpc.DialContext(ctx, pws.auth, grpc.WithInsecure(), grpc.WithBalancer(b))
		//		conn, connerr := grpc.Dial(pws.auth, grpc.WithInsecure())
		if connerr != nil {
			log.Errorf("authhost connection error: %v", connerr)
			return false, connerr
		}
		defer conn.Close()
		cli := pb.NewAuthServiceClient(conn)
		resp, err := cli.AuthToken(ctx, &pb.AuthTokenRequest{Token: token, ResourcePath: rp})
		if err != nil {
			log.Errorf("auth error: %v", err)
			return false, err
		}
		if !resp.GetValid() {
			log.Errorf("authtoken error: %s", resp.GetErr())
			return false, errors.New(resp.GetErr())
		}
		log.Infof("valid time until %s\n", time.Unix(resp.GetValidUntil()/1000, 0).Format("2006-01-02 03:04:05 PM"))
		pws.tkvalid[token.GetUser()] = resp.GetValidUntil()
		pws.rpvalid[uname] = append(pws.rpvalid[uname], rp)
	}
	return true, nil
}

func get_internal() string {
	addrs, err := net.InterfaceAddrs()
	if err != nil {
		log.Errorf("get ipAddress error:%v", err)
		return ""
	}
	for _, a := range addrs {
		if ipnet, ok := a.(*net.IPNet); ok && !ipnet.IP.IsLoopback() {
			if ipnet.IP.To4() != nil && ipnet.IP.IsGlobalUnicast() {
				return ipnet.IP.String()
			}
		}
	}
	return ""
}

func (pws *ProcWatcherService) Query(ctx context.Context, req *pb.QueryWatcherRequest) (*pb.QueryWatcherResponse, error) {
	log.Infof("handle query request: %s ,fresh %v", req.ProcName, req.Refresh)
	pws.mu.Lock()
	defer pws.mu.Unlock()
	//	var rp string
	//	host, _ := os.Hostname()
	//	rp = "/" + host + ".ecld.com/" + req.ProcName + "?action=query"
	var rp url.URL
	rp.Host = get_internal()
	rp.Path = req.ProcName
	rp.RawQuery = "action=query"
	pass, _ := pws.Auth(ctx, req.Token, strings.Replace(rp.String(), "/", "", 1))
	log.Infof("handle query request, token  passed: %v", pass)
	if !pass {
		return nil, fmt.Errorf("Insufficient Privilege")
	}
	proc, ok := pws.procs[req.ProcName]
	if req.Refresh || time.Now().Unix()-pws.lastScanTime > pws.RefreshInterval {
		log.Infof("handle query request, scan start")
		if err := pws.scan(); err != nil {
			return nil, err
		}
	}
	ret := &pb.QueryWatcherResponse{}
	if !ok {
		ret.Err = fmt.Sprintf("proc not found: %s", req.ProcName)
		return ret, nil
	}
	if _, ye := proc.Tags["port"]; !ye {
		proc.Tags["port"] = ""
	}
	stat, err := proc.process.Status(proc.Tags["port"])
	ret.Status = stat
	log.Infof("handle query request finish with：%v", ret)
	return ret, err
}

func (pws *ProcWatcherService) List(ctx context.Context, req *pb.ListRequest) (*pb.ListResponse, error) {
	log.Infof("handle list request")
	pws.mu.RLock()
	defer pws.mu.RUnlock()
	//	var rp string
	//	host, _ := os.Hostname()
	//	rp = "/" + host + ".ecld.com?action=list"
	var rp url.URL
	rp.Host = get_internal()
	rp.RawQuery = "action=list"
	pass, _ := pws.Auth(ctx, req.Token, strings.Replace(rp.String(), "/", "", 1))
	if !pass {
		return nil, fmt.Errorf("Insufficient Privilege")
	}
	var procNames []string
	for n := range pws.procs {
		procNames = append(procNames, n)
	}
	log.Infof("finished list request")
	return &pb.ListResponse{ProcName: procNames}, nil
}

func (pws *ProcWatcherService) Show(ctx context.Context, req *pb.ShowRequest) (*pb.ShowResponse, error) {
	log.Infof("handle show %s request", req.ProcName)
	pws.mu.RLock()
	defer pws.mu.RUnlock()
	//	var rp string
	//	host, _ := os.Hostname()
	//	rp = "/" + host + ".ecld.com/" + req.ProcName + "?action=show"
	var rp url.URL
	rp.Host = get_internal()
	rp.Path = req.ProcName
	rp.RawQuery = "action=show"
	pass, _ := pws.Auth(ctx, req.Token, strings.Replace(rp.String(), "/", "", 1))
	if !pass {
		return nil, fmt.Errorf("Insufficient Privilege")
	}
	proc, ok := pws.procs[req.ProcName]
	ret := &pb.ShowResponse{Commands: make(map[string]string)}
	if ok {
		for k, v := range proc.Commands {
			ret.Commands[k] = strings.Join(v, " ")
		}
	}
	log.Infof("finished show %s request", req.ProcName)
	return ret, nil
}

func (pws *ProcWatcherService) Run(req *pb.RunRequest, srv pb.WatcherService_RunServer) error {
	log.Infof("handle run request  proc name: %s , command name: %s", req.ProcName, req.CommandName)
	pws.mu.Lock()
	defer pws.mu.Unlock()
	//	var rp string
	//	host, _ := os.Hostname()
	//	rp = "/" + host + ".ecld.com/" + req.ProcName + "/" + req.CommandName + "?action=run"
	var rp url.URL
	rp.Host = get_internal()
	rp.Path = req.ProcName + "/" + req.CommandName
	rp.RawQuery = "action=run"
	pass, err := pws.Auth(context.Background(), req.Token, strings.Replace(rp.String(), "/", "", 1))
	if !pass {
		return fmt.Errorf("Insufficient Privilege")
	}
	proc, ok := pws.procs[req.ProcName]
	if !ok {
		return fmt.Errorf("no such proc: %s", req.ProcName)
	}
	cmds, ok := proc.Commands[req.CommandName]
	if !ok {
		return fmt.Errorf("no such command: %s", req.CommandName)
	}
	cmds = append(cmds, req.GetCommandArgs()...)
	cmd := exec.Command(cmds[0], cmds[1:]...)
	cmd.Env = append(cmd.Env, os.Environ()...)
	stderr, err := cmd.StderrPipe()
	stdout, err := cmd.StdoutPipe()
	output := bufio.NewReader(io.MultiReader(stderr, stdout))
	if err != nil {
		log.Errorf("run command error: %v", err)
		return err
	}
	ts := time.Now().Unix()
	if err := cmd.Start(); err != nil {
		log.Errorf("start command error: %v", err)
		return err
	}
	pid := int32(cmd.Process.Pid)
	for {
		line, err := output.ReadString('\n')
		if err == io.EOF {
			break
		} else if err != nil {
			return err
		}
		if line != "" {
			srv.Send(&pb.RunResponse{Line: []string{line}, Pid: pid, Starttime: ts})
		}
	}
	cmd.Wait()
	srv.Send(&pb.RunResponse{Finished: true, Success: cmd.ProcessState.Success()})
	return nil
}

func (pws *ProcWatcherService) scan() error {
	files, err := ioutil.ReadDir("/proc")
	if err != nil {
		log.Errorf("scan process error: %v", err)
		return err
	}
	for _, p := range pws.procs {
		p.Reset()
	}
	for _, f := range files {
		if !f.IsDir() {
			continue
		}
		re, rerr := regexp.Compile("[0-9]+")
		if rerr != nil {
			log.Errorf("parse pid error: %v", rerr)
			return rerr
		}
		if !re.MatchString(f.Name()) {
			continue
		}
		pid, err := strconv.Atoi(f.Name())
		if err != nil || pid <= 1 {
			continue
		}
		b, u, ctime, err := ParseCmdline(pid)
		if err != nil {
			continue
		}
		for _, p := range pws.procs {
			cmdline := string(b)
			if p.reg == nil {
				p.reg, err = regexp.Compile(p.ProcessPattern)
				checkErr(err)
			}
			if !p.reg.MatchString(cmdline) {
				continue
			}
			/*if p.ProcessUser != "" && p.ProcessUser == u.Username {
				err := p.process.Add(*u, ctime, pid)
				checkErr(err)
			} else {
				log.Errorf("config user is wrong,the real user is==>%s", u.Username)
			}*/
			err := p.process.Add(*u, ctime, pid)
			checkErr(err)
		}
	}
	return nil
}
